from django.shortcuts import render, redirect, get_object_or_404
from django.http import HttpResponseBadRequest
from django.urls import reverse
from django.core.paginator import Paginator
from django.db.models import Q
from django.forms import modelform_factory
from django.db.models.deletion import ProtectedError
from django.utils import timezone
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
import re
import traceback
import urllib.parse
import datetime
import threading
from concurrent.futures import ThreadPoolExecutor

from .models import Mailbox, EmailMessage, process_extraction_and_merge, ExtractionRun, ExtractionResult, AIProvider, Scene, Shipment
from core.views import _fetch_recent_emails, _select_scene_by_subject, _render_prompt, _call_ai_provider


# ==== 后台调度器（运行中心开关）====
SCHED_STATE = {
    'fetch_thread': None,
    'process_thread': None,
    'stop_event': threading.Event(),
    'running': False,
    'interval': 60,   # 秒
    'workers': 4,
    'limit': 20,
    'last_msg': '',
}


def _get_queue_stats():
    try:
        queue_pending = EmailMessage.objects.filter(status='pending').count()
        queue_running = EmailMessage.objects.filter(status='running').count()
        queue_failed = EmailMessage.objects.filter(status='failed').count()
    except Exception:
        queue_pending = queue_running = queue_failed = 0
    return {
        'queue_pending': queue_pending,
        'queue_running': queue_running,
        'queue_failed': queue_failed,
        'is_running': SCHED_STATE['running'],
    }


def _broadcast_run_center_state(extra=None):
    data = _get_queue_stats()
    if extra:
        try:
            data.update(extra)
        except Exception:
            pass
    try:
        channel_layer = get_channel_layer()
        if channel_layer:
            async_to_sync(channel_layer.group_send)(
                "run_center",
                {
                    "type": "run_center_update",
                    "data": data,
                }
            )
    except Exception:
        # 忽略广播异常
        pass


def _claim_pending_batch(batch_size: int, mailbox_id: int = None):
    from django.db import transaction
    qs = EmailMessage.objects.select_for_update(skip_locked=True).filter(status='pending')
    if mailbox_id:
        qs = qs.filter(mailbox_id=mailbox_id)
    with transaction.atomic():
        items = list(qs.order_by('id')[:batch_size])
        now = timezone.now()
        token = timezone.now().strftime('%Y%m%d%H%M%S%f')
        for e in items:
            e.status = 'running'
            e.lock_token = token
            e.locked_at = now
        if items:
            EmailMessage.objects.bulk_update(items, ['status', 'lock_token', 'locked_at'])
        if items:
            transaction.on_commit(lambda: _broadcast_run_center_state())
        return items
def _process_one_email(email_obj):
    try:
        scene = _select_scene_by_subject(email_obj)
        if not scene:
            # 无匹配场景，标记完成
            email_obj.status = 'done'
            email_obj.processed_at = timezone.now()
            email_obj.last_error = ''
            email_obj.lock_token = ''
            email_obj.locked_at = None
            email_obj.save(update_fields=['status', 'processed_at', 'last_error', 'lock_token', 'locked_at'])
            _broadcast_run_center_state()
            return True, False
        prompt = _render_prompt(scene, email_obj)
        ai_data, reasoning = _call_ai_provider(scene.ai_provider, prompt)
        result = process_extraction_and_merge(scene=scene, mailbox=email_obj.mailbox, email=email_obj, results=ai_data, reasoning=reasoning)
        email_obj.status = 'done'
        email_obj.processed_at = timezone.now()
        email_obj.last_error = ''
        email_obj.lock_token = ''
        email_obj.locked_at = None
        email_obj.save(update_fields=['status', 'processed_at', 'last_error', 'lock_token', 'locked_at'])
        _broadcast_run_center_state()
        updated_flag = False
        try:
            if isinstance(result, dict):
                updated_flag = bool(result.get('updated'))
        except Exception:
            updated_flag = False
        return True, updated_flag
    except Exception as ex:
        email_obj.status = 'failed'
        email_obj.last_error = str(ex)[:1000]
        email_obj.lock_token = ''
        email_obj.locked_at = None
        email_obj.processed_at = timezone.now()
        email_obj.save(update_fields=['status', 'last_error', 'lock_token', 'locked_at', 'processed_at'])
        _broadcast_run_center_state()
        return False, False

def _fetch_loop():
    '''
        仅负责按固定间隔拉取启用邮箱的新邮件（不做处理）
    '''
    try:
        while not SCHED_STATE['stop_event'].is_set():
            try:
                mboxes = list(Mailbox.objects.filter(enabled=True).order_by('id'))
                fetched_total = skipped_total = 0
                for mb in mboxes:
                    stats = _fetch_recent_emails(mb, limit=SCHED_STATE['limit'])
                    fetched_total += stats.get('fetched', 0)
                    skipped_total += stats.get('skipped', 0)
                SCHED_STATE['last_msg'] = f"拉取完成：新增 {fetched_total}，跳过 {skipped_total}"
                if fetched_total > 0:
                    _broadcast_run_center_state()
            except Exception as ex:
                SCHED_STATE['last_msg'] = f"拉取异常：{ex}"
            # 固定间隔执行拉取，不被处理耗时影响
            SCHED_STATE['stop_event'].wait(max(1, int(SCHED_STATE['interval'])))
    finally:
        pass


# 新增：常驻处理循环，基于线程池空闲容量持续认领并处理 pending 邮件
def _process_loop():
    from concurrent.futures import ThreadPoolExecutor
    workers = max(1, int(SCHED_STATE.get('workers') or 1))
    ok = fail = upd = 0
    futures = set()
    executor = ThreadPoolExecutor(max_workers=workers)
    try:
        while not SCHED_STATE['stop_event'].is_set():
            # 清理已完成任务，累计统计
            done_list = [f for f in list(futures) if f.done()]
            for f in done_list:
                futures.discard(f)
                try:
                    success, updated = f.result()
                    if success:
                        ok += 1
                    else:
                        fail += 1
                    if updated:
                        upd += 1
                except Exception:
                    fail += 1
            # 按线程池空闲容量认领并提交任务
            capacity = max(0, workers - len(futures))
            if capacity > 0:
                batch = _claim_pending_batch(capacity)
                if batch:
                    for email_obj in batch:
                        fut = executor.submit(_process_one_email, email_obj)
                        futures.add(fut)
                else:
                    # 无任务可处理，短暂休眠避免空转
                    SCHED_STATE['stop_event'].wait(0.5)
            else:
                # 线程池满载，短暂休眠再尝试
                SCHED_STATE['stop_event'].wait(0.1)
            # 更新处理状态提示
            SCHED_STATE['last_msg'] = f"处理队列：进行中 {len(futures)}，累计 成功 {ok} 失败 {fail} 更新 {upd}"
    finally:
        executor.shutdown(wait=False)


def run_center(request):
    """运行中心页面：默认展示所有已启用邮箱，无需选择器。"""
    enabled_mailboxes = Mailbox.objects.filter(enabled=True).order_by('name')
    ok = request.GET.get('ok')
    msg = request.GET.get('msg') or ''
    # 初始指标（用于仪表盘卡片的静态回显；后续可由前端轮询覆盖）
    try:
        queue_pending = EmailMessage.objects.filter(status='pending').count()
        queue_running = EmailMessage.objects.filter(status='running').count()
        queue_failed = EmailMessage.objects.filter(status='failed').count()
    except Exception:
        queue_pending = queue_running = queue_failed = 0
    ctx = {
        'enabled_mailboxes': enabled_mailboxes,
        'ok': ok,
        'msg': msg,
        'is_running': SCHED_STATE['running'],
        'interval': SCHED_STATE['interval'],
        'workers': SCHED_STATE['workers'],
        'limit': SCHED_STATE['limit'],
        'last_msg': SCHED_STATE['last_msg'],
        'queue_pending': queue_pending,
        'queue_running': queue_running,
        'queue_failed': queue_failed,
    }
    return render(request, 'run_center.html', ctx)


def run_start(request):
    """手动执行一轮：遍历所有启用邮箱，拉取最近邮件并按场景处理。"""
    if request.method != 'POST':
        return HttpResponseBadRequest('Only POST allowed')

    try:
        limit = int(request.POST.get('limit') or 20)
    except Exception:
        limit = 20
    limit = max(1, min(limit, 200))

    mboxes = list(Mailbox.objects.filter(enabled=True).order_by('id'))
    if not mboxes:
        return redirect('/run?ok=0&msg=' + urllib.parse.quote('没有启用的邮箱'))

    fetched_total = 0
    skipped_total = 0
    processed = 0
    errors = 0

    for mb in mboxes:
        stats = _fetch_recent_emails(mb, limit=limit)
        fetched_total += stats.get('fetched', 0)
        skipped_total += stats.get('skipped', 0)
        print(f'{mb.name}: {stats}')
        # 对新增的邮件进行抽取与合并
        for eid in stats.get('created_ids', []):
            try:
                email_obj = EmailMessage.objects.get(pk=eid)
            except EmailMessage.DoesNotExist:
                continue
            try:
                # 先按主题检查委托单号（WT14位）
                subj = (email_obj.subject or '').strip()
                po_matches = re.findall(r'WT\d{14}', subj)
                print(f'subj: {subj}')
                print(f'po_matches: {po_matches}')
                if not po_matches:
                    run = ExtractionRun.objects.create(scene=None, mailbox=mb, email=email_obj, status='success')
                    run.reasoning = '主题无委托单号'
                    run.finished_at = timezone.now()
                    run.save(update_fields=['status', 'finished_at', 'reasoning'])
                    processed += 1
                    continue

                po_no = po_matches[0]
                # 选择场景
                scene = _select_scene_by_subject(email_obj)
                if not scene:
                    continue
                print(f'{email_obj}的场景名称: {scene.name}')
                # 渲染提示词并调用 AI
                prompt = _render_prompt(scene, email_obj)
                ai_data, reasoning = _call_ai_provider(scene.ai_provider, prompt)
                # 字段映射（中文 -> 英文模型字段）
                key_map = {
                    '船名': 'vessel',
                    '航次': 'voyage',
                    '箱号': 'container_no',
                    'MBL': 'bl_no',
                    'HBL': 'hbl_no',
                    'ETD': 'etd',
                    'ETA': 'eta',
                    'ATA': 'ata',
                    '委托单号': 'po_no',
                    '业务委托号': 'po_no',
                    '提货日期': 'pickup_date',
                    'PO': 'po_no',
                }
                results = {
                    'po_no': {'value': po_no, 'confidence': 1.0, 'source': 'regex'}
                }
                for k, v in (ai_data or {}).items():
                    mk = key_map.get(k, k)
                    if mk not in scene.fields_to_extract:
                        continue
                    if mk == 'po_no':
                        # 以主题提取的委托单号为准，不被覆盖
                        continue
                    results[mk] = {'value': v or ''}
                process_extraction_and_merge(scene=scene, mailbox=mb, email=email_obj, results=results, reasoning=reasoning)
                processed += 1
            except Exception:
                print(f"{traceback.format_exc()}")
                errors += 1

    msg = f"运行完成：邮箱 {len(mboxes)} 个，新增邮件 {fetched_total}，跳过 {skipped_total}，处理 {processed}，错误 {errors}"
    _broadcast_run_center_state()
    return redirect('/run?ok=1&msg=' + urllib.parse.quote(msg))

# ================= 通用数据浏览与CRUD =================

# 支持浏览的模型映射
_BROWSE_MODEL_MAP = {
    'AIProvider': AIProvider,
    'Scene': Scene,
    'Mailbox': Mailbox,
    'EmailMessage': EmailMessage,
    'ExtractionRun': ExtractionRun,
    'ExtractionResult': ExtractionResult,
    'Shipment': Shipment,
}


def _get_model_cls(key: str):
    return _BROWSE_MODEL_MAP.get(key)


def _format_value(v):
    if v is None:
        return ''
    # 格式化日期时间为本地时区（Asia/Shanghai），并统一格式
    if isinstance(v, datetime.datetime):
        try:
            v_local = timezone.localtime(v) if timezone.is_aware(v) else timezone.make_aware(v, timezone.get_current_timezone())
            return v_local.strftime('%Y-%m-%d %H:%M:%S')
        except Exception:
            try:
                return v.strftime('%Y-%m-%d %H:%M:%S')
            except Exception:
                return str(v)
    if isinstance(v, datetime.date) and not isinstance(v, datetime.datetime):
        try:
            return v.strftime('%Y-%m-%d')
        except Exception:
            return str(v)
    s = str(v)
    return s if len(s) <= 120 else s[:117] + '...'


def browse_router(request, model_key: str = None, pk: int = None, action: str = None):
    # 选择模型(ExtractionRun)
    model_key = model_key or request.GET.get('browse_model') or 'ExtractionRun'
    if model_key not in _BROWSE_MODEL_MAP:
        model_key = 'ExtractionRun'
    model_cls = _get_model_cls(model_key)

    # 动作分派
    action = action or request.GET.get('action') or 'list'

    # 列表与搜索分页
    if action == 'list':
        q = (request.GET.get('q') or '').strip()
        # 高级筛选（仅对 ExtractionRun 生效）
        email_subject = (request.GET.get('email_subject') or '').strip()
        po_no = (request.GET.get('po_no') or '').strip()
        status = (request.GET.get('status') or '').strip()
        try:
            scene_id = int(request.GET.get('scene_id') or 0)
        except Exception:
            scene_id = 0
        try:
            mailbox_id = int(request.GET.get('mailbox_id') or 0)
        except Exception:
            mailbox_id = 0
        date_from = (request.GET.get('date_from') or '').strip()  # 期望格式 YYYY-MM-DD
        date_to = (request.GET.get('date_to') or '').strip()

        try:
            page = int(request.GET.get('page') or 1)
        except Exception:
            page = 1
        try:
            page_size = int(request.GET.get('page_size') or 10)
        except Exception:
            page_size = 10
        page_size = max(1, min(page_size, 100))

        qs = model_cls.objects.all()
        # 通用关键字搜索（当前模型的文本字段 + ID）
        if q:
            q_filter = Q()
            for f in model_cls._meta.fields:
                it = f.get_internal_type()
                if it in ('CharField', 'TextField', 'URLField', 'EmailField'):
                    q_filter |= Q(**{f.name + '__icontains': q})
            if q.isdigit():
                q_filter |= Q(**{model_cls._meta.pk.name: int(q)})
            qs = qs.filter(q_filter)
        # ExtractionRun 专属过滤
        if model_key == 'ExtractionRun':
            # 主题
            if email_subject:
                qs = qs.filter(email__subject__icontains=email_subject)
            # 状态
            if status:
                qs = qs.filter(status=status)
            # 场景/邮箱
            if scene_id:
                qs = qs.filter(scene_id=scene_id)
            if mailbox_id:
                qs = qs.filter(mailbox_id=mailbox_id)
            # 时间范围（使用 email.date 为主，缺失则不限制）
            if date_from:
                try:
                    qs = qs.filter(Q(email__date__date__gte=date_from) | Q(email__date__isnull=True))
                except Exception:
                    pass
            if date_to:
                try:
                    qs = qs.filter(Q(email__date__date__lte=date_to) | Q(email__date__isnull=True))
                except Exception:
                    pass
            # PO 号（联查 ExtractionResult）
            if po_no:
                qs = qs.filter(extractionresult__field='po_no', extractionresult__value__icontains=po_no)
            qs = qs.distinct()

        qs = qs.order_by('-' + model_cls._meta.pk.name)
        paginator = Paginator(qs, page_size)
        page_obj = paginator.get_page(page)

        # 列表数据与字段
        objs = list(page_obj.object_list)
        field_names = [f.name for f in model_cls._meta.fields]
        rows = []
        pk_name = model_cls._meta.pk.name
        for o in objs:
            row = {f: _format_value(getattr(o, f)) for f in field_names}
            row['pk'] = getattr(o, pk_name, o.pk)
            rows.append(row)

        # 场景与邮箱选项
        scenes = Scene.objects.all()
        mailboxes = Mailbox.objects.all()
        statuses = ['pending', 'running', 'success', 'failed', 'done']

        # 回填查询参数
        params = {
            'q': q,
            'email_subject': email_subject,
            'po_no': po_no,
            'status': status,
            'scene_id': scene_id,
            'mailbox_id': mailbox_id,
            'date_from': date_from,
            'date_to': date_to,
            'page_size': page_size,
        }
        query_string = '&'.join([f"{k}={urllib.parse.quote(str(v))}" for k, v in params.items() if v])

        return render(request, 'browse.html', {
            'mode': 'list',
            'browse_models': list(_BROWSE_MODEL_MAP.keys()),
            'browse_model_key': model_key,
            'field_names': field_names,
            'rows': rows,
            'page_obj': page_obj,
            'page_size': page_size,
            'q': q,
            'email_subject': email_subject,
            'po_no': po_no,
            'status': status,
            'scene_id': scene_id,
            'mailbox_id': mailbox_id,
            'date_from': date_from,
            'date_to': date_to,
            'scenes': scenes,
            'mailboxes': mailboxes,
            'statuses': statuses,
            'query_string': query_string,
        })

    # 创建
    if action == 'create':
        pk_name = model_cls._meta.pk.name
        include_fields = [f.name for f in model_cls._meta.fields if f.editable and f.name != pk_name]
        FormCls = modelform_factory(model_cls, fields=include_fields)
        if request.method == 'POST':
            form = FormCls(request.POST)
            if form.is_valid():
                obj = form.save()
                return redirect(reverse('browse_detail', kwargs={'model_key': model_key, 'pk': obj.pk}))
        else:
            form = FormCls()
        return render(request, 'browse.html', {
            'mode': 'form',
            'form_mode': 'create',
            'browse_models': ['ExtractionRun'],
            'browse_model_key': model_key,
            'form': form,
        })

    # 详情
    if action == 'detail':
        obj = get_object_or_404(model_cls, pk=pk)
        field_names = [f.name for f in model_cls._meta.fields]
        fields = [(f, _format_value(getattr(obj, f))) for f in field_names]

        ctx = {
            'mode': 'detail',
            'browse_models': list(_BROWSE_MODEL_MAP.keys()),
            'browse_model_key': model_key,
            'object': obj,
            'fields': fields,
        }

        # 为 ExtractionRun 模型补充 EmailMessage 详情与抽取结果列表（不修改模板）
        if model_key == 'ExtractionRun':
            try:
                email_obj = obj.email
            except Exception:
                email_obj = None
            if email_obj:
                email_field_names = [f.name for f in EmailMessage._meta.fields]
                email_details = [(f, _format_value(getattr(email_obj, f))) for f in email_field_names]
            else:
                email_details = []
            results = list(ExtractionResult.objects.filter(run=obj).order_by('id'))
            ctx.update({
                'email_details': email_details,
                'results': results,
            })

        return render(request, 'browse.html', ctx)

    # 编辑
    if action == 'edit':
        obj = get_object_or_404(model_cls, pk=pk)
        pk_name = model_cls._meta.pk.name
        include_fields = [f.name for f in model_cls._meta.fields if f.editable and f.name != pk_name]
        FormCls = modelform_factory(model_cls, fields=include_fields)
        if request.method == 'POST':
            form = FormCls(request.POST, instance=obj)
            if form.is_valid():
                obj = form.save()
                return redirect(reverse('browse_detail', kwargs={'model_key': model_key, 'pk': obj.pk}))
        else:
            form = FormCls(instance=obj)
        return render(request, 'browse.html', {
            'mode': 'form',
            'form_mode': 'edit',
            'browse_models': ['ExtractionRun'],
            'browse_model_key': model_key,
            'form': form,
            'object': obj,
        })

    # 删除
    if action == 'delete':
        obj = get_object_or_404(model_cls, pk=pk)
        if request.method == 'POST':
            try:
                obj.delete()
                return redirect(reverse('browse_model', kwargs={'model_key': model_key}))
            except ProtectedError:
                return HttpResponseBadRequest('该记录被引用，无法删除（ProtectedError）')
        return render(request, 'browse.html', {
            'mode': 'delete',
            'browse_models': list(_BROWSE_MODEL_MAP.keys()),
            'browse_model_key': model_key,
            'object': obj,
        })

    return HttpResponseBadRequest('Unsupported action')


def run_daemon_start(request):
    """启动后台定时器：按前端配置的间隔与线程数循环拉取并处理。"""
    if request.method != 'POST':
        return HttpResponseBadRequest('Only POST allowed')
    try:
        interval = int(request.POST.get('interval') or SCHED_STATE['interval'])
        workers = int(request.POST.get('workers') or SCHED_STATE['workers'])
        limit = int(request.POST.get('limit') or SCHED_STATE['limit'])
    except Exception:
        return HttpResponseBadRequest('参数不合法')
    interval = max(5, min(interval, 3600))
    workers = max(1, min(workers, 32))
    limit = max(1, min(limit, 200))

    # 判断是否已在运行（任一线程存活即视为运行中）
    ft = SCHED_STATE.get('fetch_thread')
    pt = SCHED_STATE.get('process_thread')
    if SCHED_STATE['running'] and ((ft and ft.is_alive()) or (pt and pt.is_alive())):
        return redirect('/run?ok=1&msg=' + urllib.parse.quote('后台已在运行'))

    # 应用新配置并启动两个解耦线程
    SCHED_STATE['interval'] = interval
    SCHED_STATE['workers'] = workers
    SCHED_STATE['limit'] = limit
    SCHED_STATE['stop_event'].clear()
    SCHED_STATE['running'] = True

    ft = threading.Thread(target=_fetch_loop, name='mailai-fetcher', daemon=True)
    pt = threading.Thread(target=_process_loop, name='mailai-processor', daemon=True)
    SCHED_STATE['fetch_thread'] = ft
    SCHED_STATE['process_thread'] = pt
    ft.start()
    pt.start()
    _broadcast_run_center_state({'is_running': True})
    return redirect('/run?ok=1&msg=' + urllib.parse.quote(f'已启动（解耦合）：拉取每 {interval}s，处理线程 {workers}，每邮箱拉取 {limit} 封'))


def run_daemon_stop(request):
    """停止后台定时器。"""
    if request.method != 'POST':
        return HttpResponseBadRequest('Only POST allowed')
    any_alive = False
    for key in ('fetch_thread', 'process_thread', 'thread'):
        t = SCHED_STATE.get(key)
        if t and hasattr(t, 'is_alive') and t.is_alive():
            any_alive = True
    if any_alive:
        SCHED_STATE['stop_event'].set()
        SCHED_STATE['running'] = False
        _broadcast_run_center_state({'is_running': False})
        # 非阻塞地等待线程自行退出
    return redirect('/run?ok=1&msg=' + urllib.parse.quote('已发送停止指令，稍后刷新查看状态'))